package com.kd.crawler.parser.thread;

import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.Queue;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.kd.crawler.common.jms.ActiveMQProducer;
import com.kd.crawler.common.utils.IJedisPool;
import com.kd.crawler.common.utils.MD5Util;
import com.kd.crawler.common.utils.PropertiesUtils;
import com.kd.crawler.entity.CrawlerEntry;

/**
 * @author Manfred
 *
 */
public class URLFilterQueue implements Runnable{
	private Logger logger=Logger.getLogger(URLFilterQueue.class);
	Properties properties = PropertiesUtils.getProperties("activeMQ.properties");
	private static Queue<CrawlerEntry> urlMd5Queue = new LinkedList<CrawlerEntry>();
	
	public void run() {
		while(true){
			CrawlerEntry message=poll();	
			//CrawlerEntry entry=JSON.parseObject(message, CrawlerEntry.class);
			String urlMD5=MD5Util.MD5(message.getUrl());
			long flag=IJedisPool.sadd("KD_CRAWL_URL_MD5_SET", urlMD5);
			if(!(flag>0)){
				continue;
			}			
			logger.info("正在发送入口信息："+message);
			//MQMessageSender.offerQueueMessage(properties.getProperty("QUEUE_CRAWL_ENTRY_MESSAGE").trim(), JSON.toJSONString(message));
			ActiveMQProducer.sendQueueMessage(properties.getProperty("QUEUE_CRAWL_ENTRY_MESSAGE").trim(), JSON.toJSONString(message));
//			EntryMessageSender.offer(JSON.toJSONString(entry));
		}
		
	}

	public static synchronized CrawlerEntry poll() {
		CrawlerEntry message = null;
		while (message == null) {
			message = urlMd5Queue.poll();
			if (message == null) {
				try {
					URLFilterQueue.class.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		return message;
	}
	public static synchronized void offer(CrawlerEntry crawlerEntry) {
		urlMd5Queue.offer(crawlerEntry);
		URLFilterQueue.class.notify();
	}
	
	public static synchronized void offer(List<CrawlerEntry> crawlerEntries) {
		for(CrawlerEntry item:crawlerEntries){
			urlMd5Queue.offer(item);
		}
		URLFilterQueue.class.notify();
	}

}
